package com.ruoyi.xdata.task;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson2.JSON;
import com.ruoyi.common.constant.HttpStatus;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.core.page.TableDataInfo;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.system.service.ISysConfigService;
import com.ruoyi.util.DbUtil;
import com.ruoyi.util.HttpUtil;
import com.ruoyi.xdata.domain.XdataLog;
import com.ruoyi.xdata.domain.XdataModel;
import com.ruoyi.xdata.domain.XdataPipe;
import com.ruoyi.xdata.domain.XdataVo;
import com.ruoyi.xdata.mapper.XdataModelMapper;
import com.ruoyi.xdata.service.IXdataLogService;
import com.ruoyi.xdata.service.IXdataPipeService;

/**
 * 数据推送的定时任务
 * @author Administrator
 *
 */
@Component("pushTask")
public class PushTaskJob {
	private static final Logger log = LoggerFactory.getLogger(PushTaskJob.class);
	private static int PUSH_ROWS = 100;
	@Autowired	 
    private DataSource dataSource; 
	@Autowired	 
	private XdataModelMapper xdataModelMapper;
    @Autowired
    private IXdataPipeService xdataPipeService;
    @Autowired
    private ISysConfigService configService;	
    @Autowired
    private IXdataLogService xdataLogService;

	/**
	 * 向主服务器推送数据任务
	 */
	public void push(String table, String url) {
		XdataModel model = xdataModelMapper.selectXdataModelByTable(table);
		if(model == null) {
			model = xdataModelMapper.selectXdataModelById(Long.parseLong(table));
		}
		if(model == null) {
			log.error("数据模型 {} 不存在", table);
			return;
		}


		XdataPipe pipe = getPipe(model, "R");
		if(pipe == null) {
			log.error("未定义数据通道,model {}, client {} ", table, configService.selectConfigByKey("push.server.id"));
			return;
		}
				
		String token = getToken(url);
		if(token == null) {
			return;
		}
		

		Date lt = pipe.getLastTime();
		if(lt == null) {
			lt = new Date(1l);
		}
		
		int p = 0;
		Map<String,Object> params = new HashMap<>();
		params.put("limit", PUSH_ROWS);
		params.put("_last", lt);
		
		DbUtil read = null;
		try {
			read = new DbUtil(dataSource.getConnection());
			while(true) {
				params.put("offset", p * PUSH_ROWS);
				List<Map> list = read.query(pipe.getReadSql(), params);
				
				if(list == null ||list.size() == 0) break;

				XdataVo vo = new XdataVo();
				vo.setTable(model.getTables());
				vo.setToken(token);
				List<Map<String, Object>> rows = new ArrayList<Map<String,Object>>();
				for(Map data : list) {
					rows.add(data);
				}
				vo.setRows(rows);
				AjaxResult ret = post(url + "/xdata/data/push", JSON.toJSONString(vo));
				
				if(ret == null) {
		    		log.error("push data error {} {} ", vo, ret.get(AjaxResult.CODE_TAG));
				}
			}		
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	/**
	 * 从主服务器拉数据任务
	 */
	public void pull(String table, String url) {
		XdataModel model = xdataModelMapper.selectXdataModelByTable(table);
		if(model == null) {
			model = xdataModelMapper.selectXdataModelById(Long.parseLong(table));
		}
		if(model == null) {
			log.error("数据模型 {} 不存在", table);
			return;
		}


		XdataPipe pipe = getPipe(model, "W");
		if(pipe == null) {
			log.error("未定义数据通道,model {}, client {} ", table, configService.selectConfigByKey("push.server.id"));
			return;
		}
		
		String token = getToken(url);
		if(token == null) {
			return;
		}
		
		Date lt = pipe.getLastTime();
		if(lt == null) {
			lt = new Date(1l);
		}
		XdataVo vo = new XdataVo();
		vo.setTable(model.getTables());
		vo.setToken(token);
		vo.setLimit(PUSH_ROWS);
		
		int p = 0;
		int total = 0;
		int success = 0;
		XdataLog xdataLog = new XdataLog();
		xdataLog.setStartTime(new Date());
		xdataLog.setPipeId(pipe.getId());
		DbUtil writer = null;
		try {
			writer = new DbUtil(dataSource.getConnection());
		} catch (SQLException e1) {
			e1.printStackTrace();
		}
	    		     
		while(true) {
			vo.setOffset(p * PUSH_ROWS);

			TableDataInfo ret = get(url + "/xdata/data/list", vo.toUrl());
			if(ret == null) {
	    		log.error("pull data error {} {} ", url, vo.toUrl());
			}
			if(ret.getRows() == null || ret.getRows().size() == 0) {
				break;
			}

			List<Map> list = (List<Map>)ret.getRows();
			
			for(Map data : list) {
				try {
			    	if(writer.find(pipe.getExistSql(), data) == null) {
			    		writer.exec(pipe.getInsertSql(), data);
					}else {
						writer.exec(pipe.getUpdateSql(), data);
					}
			    	success ++;
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					
					xdataLog.setId(null);
					xdataLog.setEndTime(new Date());
					xdataLog.setResult("0");
					xdataLog.setError(e.getMessage());
					xdataLogService.insertXdataLog(xdataLog);
				}
			}
			total += list.size();
			if(ret.getTotal() <= total) {
				break;
			}
			p ++;
		}
		pipe.setLastTime(new Date());
		pipe.setLastResult(0l + total);
		xdataPipeService.updateXdataPipe(pipe);

		xdataLog.setId(null);
		xdataLog.setEndTime(new Date());
		xdataLog.setResult("1");
		xdataLog.setRows(success + 0l);
		xdataLogService.insertXdataLog(xdataLog);
	}
	
	
	/**
	 * 发起POST请求
	 * @param url
	 * @param json
	 * @return
	 */
	public AjaxResult post(String url, String json) {
		String resp = null;
		if(url.substring(0,5).toLowerCase().equals("https")) {
			resp = HttpUtil.sendSSLPost(url, json);
		}else {
			resp = HttpUtil.sendPost(url, json);
		}
		
		if(StringUtils.isEmpty(resp)) {
    		log.error("request error:url {}", url);
			return null;
		}
		AjaxResult ret = JSON.parseObject(resp, AjaxResult.class);
		if(ret == null) {
    		log.error("parse response error  {}", resp);
			return null;
		}
		
		int code = Integer.parseInt(ret.get(AjaxResult.CODE_TAG).toString());
    	if(code != HttpStatus.SUCCESS) {
    		log.error("request operation error {} {} ", code, ret.get(AjaxResult.MSG_TAG));
    		return null;
    	}
    	
    	return ret;
	}

	/**
	 * 发起POST请求
	 * @param url
	 * @param json
	 * @return
	 */
	public TableDataInfo get(String url, String json) {
		String resp = null;
		if(url.substring(0,5).toLowerCase().equals("https")) {
			resp = HttpUtil.sendSSLGet(url, json);
		}else {
			resp = HttpUtil.sendGet(url, json);
		}
		
		if(StringUtils.isEmpty(resp)) {
    		log.error("request error:url {}", url);
			return null;
		}
		TableDataInfo ret = JSON.parseObject(resp, TableDataInfo.class);
		if(ret == null) {
    		log.error("parse response error  {}", resp);
			return null;
		}
		
    	if(ret.getCode() != HttpStatus.SUCCESS) {
    		log.error("request operation error {} ", ret.getCode());
    		return null;
    	}
    	
    	return ret;
	}
	
	/**
	 * 获取令牌
	 * @param url
	 * @return
	 */
	public String getToken(String url) {
		Map data = new HashMap<String, String>();
		data.put("code", configService.selectConfigByKey("push.client.code"));
		data.put("secret", configService.selectConfigByKey("push.client.secret"));
		
		AjaxResult ret = post(url + "/xdata/data/token", JSON.toJSONString(data));
		if(ret == null) {
    		log.error("request push token error {} {} ", ret.get(AjaxResult.CODE_TAG), ret.get(AjaxResult.MSG_TAG));
			return null;
		}
				
    	return ret.get(AjaxResult.MSG_TAG).toString();
	}
	
	 /**
     * 查询管道信息
     * @param model
     * @param mode
     * @return
     */
    private XdataPipe getPipe(XdataModel model, String mode) {
    	String clientId = configService.selectConfigByKey("push.server.id");
    	if(StringUtils.isEmpty(clientId)) {
    		return null;
    	}
    	    	
    	XdataPipe pc = new XdataPipe();
    	pc.setClientId(Long.parseLong(clientId));
    	pc.setModelId(model.getId());
    	pc.setMode(mode.toLowerCase());
    	
    	List<XdataPipe> ps = xdataPipeService.selectXdataPipeList(pc);
    	if(ps == null || ps.size() == 0) {
    		return null;
    	}   
    	return ps.get(0);
    }
}
